package org.example.flink.operation;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * @Description :
 * @School:优极限学堂
 * @Official-Website: http://www.yjxxt.com
 * @Teacher:李毅大帝
 * @Mail:863159469@qq.com
 */
public class Hello10TransformationUnionConnect {
    public static void main(String[] args) throws Exception {
        //运行环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(2);
        //获取数据流 (类型相同)
        // DataStreamSource<Integer> source1 = environment.fromElements(1, 3, 5, 7, 9);
        // DataStreamSource<Integer> source2 = environment.fromElements(2, 4, 6, 8, 9);
        // source1.union(source2).print();

        //获取数据流--火警传感器[字符串]--温度传感器[数字]
        DataStreamSource<String> source1 = environment.fromElements("yes", "yes", "no", "yes");
        DataStreamSource<Double> source2 = environment.fromElements(33.1, 33.6, 50.2, 33.3);
        ConnectedStreams<String, Double> connect = source1.connect(source2);

        connect.map(new CoMapFunction<String, Double, String>() {
            @Override
            public String map1(String value) throws Exception {
                if ("yes".equals(value)) {
                    return "[" + value + "]火警传感器[安全]";
                }
                return "[" + value + "]火警传感器[危险]";
            }

            @Override
            public String map2(Double value) throws Exception {
                if (value <= 50) {
                    return "[" + value + "]温度传感器[安全]";
                }
                return "[" + value + "]温度传感器[危险]";
            }
        }).print();

        //运行环境
        environment.execute();
    }
}
